package com.fengplus.demo.controller;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.UUID;

@RestController
@RequestMapping("/provider")
@Slf4j
public class ProviderController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // Hello模式
    @GetMapping("/a")
    public String providerA() {
        String exchange = "";
        String routeKey = "queue.a";
        String msg = "来自Provider A：" + UUID.randomUUID().toString();
        rabbitTemplate.convertAndSend(exchange, routeKey, msg);
        return "SUCCESS A";
    }

    // Work模式
    @GetMapping("/b")
    public String providerB() {
        String exchange = "";
        String routeKey = "queue.b";

        for (int i = 0; i < 20; i++) {
            String msg = "来自Provider B + " + i + " ：" + UUID.randomUUID().toString();
            rabbitTemplate.convertAndSend(exchange, routeKey, msg);
        }

        return "SUCCESS B";
    }

    // Publish/Subscribe模式
    @GetMapping("/c")
    public String providerC() {
        String exchange = "exchange.c";
        String routeKey = "";
        String msg = "来自Provider C：" + UUID.randomUUID().toString();
        rabbitTemplate.convertAndSend(exchange, routeKey, msg);
        return "SUCCESS C";
    }

    // Routing模式
    @GetMapping("/d")
    public String providerD(String route) {
        String exchange = "exchange.d";
        String routeKey = route == null ? "route.d" : route;
        String msg = "来自Provider D：" + UUID.randomUUID().toString();
        rabbitTemplate.convertAndSend(exchange, routeKey, msg);
        return "route可输入值：route.d.info route.d.error route.d.warning";
    }

    // Topic模式
    @GetMapping("/e")
    public String providerE(String route) {
        String exchange = "exchange.e";
        String routeKey = route == null ? "route.e" : route;
        String msg = "来自Provider E：" + UUID.randomUUID().toString();
        rabbitTemplate.convertAndSend(exchange, routeKey, msg);
        return "route可输入值：route.e.info route.e.error route.e.warning";
    }

    // RPC模式
    @GetMapping("/f")
    public void providerF() {
        String exchange = "exchange.f";
        String routeKey = "route.f.rpc";
        String msg = "来自Provider F：" + UUID.randomUUID().toString();
        rabbitTemplate.convertAndSend(routeKey, (Object) msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties p = message.getMessageProperties();
                p.setReplyTo("route.f.reply");
                String correlationId = UUID.randomUUID().toString();
                log.info("Provider发出的消息 -> correlationId:{}", correlationId);
                log.info("Provider发出的消息 -> msg:{}", msg);
                p.setCorrelationId(correlationId);
                return message;
            }
        });
    }

    @RabbitListener(queues = "route.f.reply")
    public void ConsumerFReply(String msg,
                               @Header(AmqpHeaders.CORRELATION_ID) String correlationId,
                               @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                               Channel channel) throws IOException {
        try {
            log.info("Reply收到的消息 -> msg:{}", msg);
            log.info("Reply收到的消息 -> correlationId:{}", correlationId);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
//            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
//            channel.basicNack(deliveryTag, false, false);
        }
    }

    // Confirm模式
    @GetMapping("/g")
    public String providerG() {
        String exchange = "exchange.g";
        String routeKey = "route.g";
        String msg = "来自Provider G：" + UUID.randomUUID().toString();;
//        rabbitTemplate.convertAndSend(exchange, routeKey, msg);
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        log.info("callbackSender UUID: " + correlationData.getId());
        this.rabbitTemplate.convertAndSend(exchange, routeKey, msg, correlationData);
        rabbitTemplate.setConfirmCallback((a, b, c) -> {
            log.info("==============");
            System.out.println(a);
            System.out.println(b);
            System.out.println(c);
        });
        return "SUCCESS G";
    }
}
